package main

import (
	"bytes"
	"flag"
	"fmt"
	"math/rand"
	"os"
	"os/exec"
	"runtime"
	"runtime/debug"
	"strconv"

	//"strconv"
	"strings"
	"sync"
	"time"
)

var (
	grow = flag.Int("grow", 0, "Buffer pre-alloc size")
	docSize = flag.Int("docsize", 0, "single write size")
	ns = flag.Int("ns", 1, "number of ns to process in parallel")
	repeat = flag.Int("repeat", 10, "repeat number of buf.Read()")
	size = flag.Int("size", 64*1024*1024, "total table size")
	reset = flag.Bool("reset", false, "whether reset bytes.Buffer after surpass the size threshold")
	free = flag.Bool("free", false, "whether call debug.FreeOSMeory")

)

func toMegaBytes(bytes uint64) float64 {
	return float64(bytes) / 1024 / 1024
}

//https://blog.csdn.net/pengpengzhou/article/details/106901368
/*
HeapObjects：堆中已经分配的对象总数，GC内存回收后HeapObjects取值相应减小。
HeapAlloc:  堆中已经分配给对象的字节数，GC内存回收后HeapAlloc取值相应减小。
TotalAlloc:  堆中已经分配给对象的总的累计字节数，只增不减，GC内存回收后也不减小。
HeapSys: 从操作系统为堆申请到的字节数。
HeapIdle: 堆的闲置区间，包括已经归还给操作系统的物理字节数（HeapReleased）
HeapReleased: 已经归还给操作系统的物理字节数，是HeapIdle的子集。
*/
func traceMemStats() {
	var ms runtime.MemStats
	runtime.ReadMemStats(&ms)
	var result = make([]float64, 6)
	result[0] = float64(ms.HeapObjects)
	result[1] = toMegaBytes(ms.HeapAlloc)
	result[2] = toMegaBytes(ms.TotalAlloc)
	result[3] = toMegaBytes(ms.HeapSys)
	result[4] = toMegaBytes(ms.HeapIdle)
	result[5] = toMegaBytes(ms.HeapReleased)
	//result[6] = toMegaBytes(ms.HeapIdle - ms.HeapReleased)
 
	//fmt.Printf("%d\t", time.Now().Unix())
	for _, v := range result {
		fmt.Printf("%.2f\t", v)
	}
	fmt.Printf("\n")
}

func printMetrics(len1, cap1, buf_cap, docSizeSum int) {
	fmt.Printf("%v,%v,%v,%v,", len1, cap1, buf_cap, docSizeSum)
	cmdStr := "cat /proc/" + fmt.Sprintf("%d", os.Getpid()) + "/status | grep RSS | awk '{print $2}'" 
	cmd := exec.Command("/bin/bash", "-c", cmdStr)
	var res []byte
	var err error
	if res, err = cmd.Output(); err != nil {
		fmt.Printf("err: %v\n", err)
		panic("cmd failed")
	}
	fmt.Print(string(res))
}



func processOneMemfile(i int, wg *sync.WaitGroup) {
	buf := new(bytes.Buffer)
	if *grow != 0 {
		buf.Grow(*grow)
	}

	var kk = 0

	//fmt.Printf("heapObj\theapAlloc\ttotalAlloc\theapSys\theapIdle\theapReleased\n")
	fmt.Printf("len,cap_buf,buf_cap,total-write,RSS\n")
	var err error
	docSizeSum := 0
	for {
		if *docSize == 0 {
			docSize1 := 1024+rand.Intn(10484736) // 没有设置种子，可以使得不同配置使用相同的随机序列
			docSizeSum += docSize1
			_, err = buf.Write([]byte(strings.Repeat("a", docSize1))) // 1K~10M
		} else {
			docSizeSum += *docSize
			_, err = buf.Write([]byte(strings.Repeat("a", *docSize)))
		}
		if err != nil {
			panic("buf write failed")
		}
		//traceMemStats()
		if buf.Len() >= *size {
			//traceMemStats()
			printMetrics(len(buf.Bytes()) / 1024, cap(buf.Bytes()) / 1024, buf.Cap() / 1024, docSizeSum/1024)

			bufToUpload := make([]byte, *size)
			// 这里读了后, buf.Cap 似乎会shrink ，但是程序的RSS 并不会缩小
			buf.Read(bufToUpload)

			// 首先测试在没有写 buf=new(bytes.Buffer)时，go 1.16 会不会让RSS 自降？
			// 然后再写上 buf=new(bytes.Buffer)

			// 不会立刻造成 buf所指内存回收 ??
			//buf = new(bytes.Buffer)

			// for {
			// 	cmdStr := "cat /proc/" + fmt.Sprintf("%d", os.Getpid()) + "/status | grep RSS | awk '{print $2}'" 
			// 	cmd := exec.Command("/bin/bash", "-c", cmdStr)
			// 	var res []byte
			// 	var err error
			// 	if res, err = cmd.Output(); err != nil {
			// 		fmt.Printf("err: %v\n", err)
			// 		panic("cmd failed")
			// 	}
			// 	time.Sleep(2*time.Second)

			// 	fmt.Print(time.Now().Format("20060102.150405"), " getting rss:", string(res))
				
			// 	a, err := strconv.Atoi(string(res[:len(res)-1]))
			// 	if err != nil {
			// 		panic("Atoi error")
			// 	}
			// 	if a < 200000 {
			// 		break
			// 	}
			// }

			//buf = new(bytes.Buffer)

			if *free {
				debug.FreeOSMemory()
				fmt.Printf("after free\n")
				printMetrics(len(buf.Bytes()) / 1024, cap(buf.Bytes()) / 1024, buf.Cap() / 1024, docSizeSum/1024)
			}

			// 指针互换能不能触发已经读过的bytes 部分回收呢？让老指针指向新对象，这样老对象会被回收？

			// 验证在这里进行Reset 能否减轻内存开销
			if *reset {
				buf.Reset()
				time.Sleep(3*time.Second)
				printMetrics(len(buf.Bytes()) / 1024, cap(buf.Bytes()) / 1024, buf.Cap() / 1024, docSizeSum/1024)
			}

			kk++

			if kk < *repeat {
				//fmt.Printf("continue, kk=%d\n", kk)
				continue
			}

			//fmt.Printf("thread %v 64M reached. exit\n", i)
			wg.Done()
			return
		}

		printMetrics(len(buf.Bytes()) / 1024, cap(buf.Bytes()) / 1024, buf.Cap() / 1024, docSizeSum/1024)
			
		//time.Sleep(time.Second * 1)
	}
}

/*
bytes.Buffer Reset之后，如果再写入新的数据，如果数据的长度没有超过Reset之前缓冲区的长度，那么Buffer内部不会重新开辟内存，也就是说，
写入的数据会覆盖之前的数据。
在本例中，这里【之前的数据】就是那个 b ， 很明显它的第一个字节已经被修改了。
*/

func testBytesBuffer() {
    a := []byte{'1', '2'}
    buf := new(bytes.Buffer)
    buf.Write(a)
	fmt.Printf("first, buf.Len=%d, buf.Cap=%d\n", buf.Len(), buf.Cap())
    b := buf.Bytes()
    fmt.Println(b)
    buf.Reset()
    c := []byte{'3'}
    buf.Write(c)
    fmt.Printf("after 1st reset, write one byte, %v\n", b) // '3' 覆盖了 '1'

	buf.Reset() // 导致 len 为0，意味着原来的数据再也读不到了？
	fmt.Printf("after reset, b=%v, buf.Bytes=%v, buf.Len=%d, buf.Cap=%d\n", b, buf.Bytes(), buf.Len(), buf.Cap())
	buf.Write([]byte("abcdefg"))
	b2 := buf.Bytes()
	fmt.Printf("after second reset, write many data, %v, b2=%v, " +
		"len(b)=%d, cap(b)=%d, len(b2)=%d, cap(b2)=%d\n", b,b2, len(b), cap(b), len(b2), cap(b2))
}

var data = make([]byte, 10000)
var bufferPool = sync.Pool{
	New: func() interface{} {
		return &bytes.Buffer{}
	},
}
// 减少在堆上的分配，缓解 GC 压力
/*
通过以上的解读，我们可以看到，Get方法并不会对获取到的对象值做任何的保证，因为放入本地池中的值有可能会在任何时候被删除，但是不通知调用者。
放入共享池中的值有可能被其他的goroutine偷走。 所以对象池比较适合用来存储一些临时切状态无关的数据，但是不适合用来存储数据库连接的实例，
因为存入对象池重的值有可能会在垃圾回收时被删除掉，这违反了数据库连接池建立的初衷。

根据上面的说法，Golang的对象池严格意义上来说是一个临时的对象池，适用于储存一些会在goroutine间分享的临时对象。主要作用是减少GC，提高性能。
在Golang中最常见的使用场景是fmt包中的输出缓冲区

那么这个池子的目的是什么呢？其实一句话就可以说明白，就是为了复用已经使用过的对象，来达到优化内存使用和回收的目的。
说白了，一开始这个池子会初始化一些对象供你使用，如果不够了呢，自己会通过new产生一些，当你放回去了之后这些对象会被别人进行复用，
当对象特别大并且使用非常频繁的时候可以大大的减少对象的创建和回收的时间。
链接：https://www.jianshu.com/p/8fbbf6c012b2

*/
func testBytesbufferWithSyncPool() {
	for n := 0; n < 5; n++ {
		buf := bufferPool.Get().(*bytes.Buffer)
		buf.Write(data)
		fmt.Printf("len=%d, cap=%d\n", buf.Len(), buf.Cap())
		buf.Reset() // Reset 导致没有unread 数据， Len()=0
		fmt.Printf("\tafter Reset, len=%d, cap=%d\n", buf.Len(), buf.Cap())
		bufferPool.Put(buf)
	}
}

func reproduceWorkerRSSHigh() {
	flag.Parse()
	var wg sync.WaitGroup
	wg.Add(*ns)

	for i:=0; i<*ns; i++ {
		go processOneMemfile(i, &wg)
	}

	wg.Wait()

	//等RSS是否会降下来
	for {
		cmdStr := "cat /proc/" + fmt.Sprintf("%d", os.Getpid()) + "/status | grep RSS | awk '{print $2}'" 
		cmd := exec.Command("/bin/bash", "-c", cmdStr)
		var res []byte
		var err error
		if res, err = cmd.Output(); err != nil {
			fmt.Printf("err: %v\n", err)
			panic("cmd failed")
		}
		time.Sleep(2*time.Second)

		fmt.Print(time.Now().Format("20060102.150405"), " getting rss:", string(res))
		
		a, err := strconv.Atoi(string(res[:len(res)-1]))
		if err != nil {
			panic("Atoi error")
		}
		if a < 400000 {
			break
		}
	}
}

func tBytes() {
	//testBytesBuffer()
	//testBytesbufferWithSyncPool()

	// go func (d time.Duration)  {
	// 	tick := time.Tick(d)
    // 	for _ = range tick {
	// 		fmt.Println("ready to free ...")
    //     	debug.FreeOSMemory()
    // 	}
	// }(10 * time.Second)

	reproduceWorkerRSSHigh()
}